package dyyx.util;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.common.Node;

import dyyx.dto.ConsumerGroupDTO;
import dyyx.dto.ConsumerGroupsInfo;
import dyyx.dto.KafkaBroker;
import dyyx.dto.KafkaClient;
import dyyx.dto.KafkaClusterInfo;
import dyyx.dto.KafkaOffsetInfo;
import dyyx.dto.KafkaTopicReplicaSummaryInfo;
import dyyx.dto.WebResult;

public abstract class KafkaViewWebUtil {

	private static final Set<String> OPS = new HashSet<>();

	private static final String OP_CONNECT = "connect";
	private static final String OP_DIS_CONNECT = "disConnect";
	private static final String OP_CLIENT_INFO = "clientInfo";
	
	private static final String CONFIG_TYPE_BROKER = "broker";
	private static final String CONFIG_TYPE_TOPIC = "topic";

	
	private static final Set<String> CONFIG_TYPES = new HashSet<>();


	static {

		OPS.add(OP_CONNECT);
		OPS.add(OP_DIS_CONNECT);
		OPS.add(OP_CLIENT_INFO);
		
		CONFIG_TYPES.add(CONFIG_TYPE_BROKER);
		CONFIG_TYPES.add(CONFIG_TYPE_TOPIC);

		
	}
	
	
	public static void run(HttpServletRequest request, WebResult webResult) throws Exception {
		try {
			run_(request, webResult);
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = webResult.op + " error," + e;
		}
	}
	
	
	private static void run_(HttpServletRequest request, WebResult webResult) throws Exception {
		String op = WebUtil.get(request, "op");
		String connstr = WebUtil.get(request, "connstr");
		

		if (StringUtils.isBlank(op)) {
			webResult.msg = "op blank";
			return;
		}
		if (!OPS.contains(op)) {
			webResult.msg = "op error";
			webResult.error = true;
			return;
		}
		webResult.op = op;
		

		if (OP_CONNECT.equals(op)) {

			if (StringUtils.isBlank(connstr)) {
				webResult.msg = "connstr blank";
				return;
			}

			KafkaUtil.connect(connstr);

			webResult.msg = op + " success,connstr=" + connstr;

			return;
		}

		if (OP_DIS_CONNECT.equals(op)) {

			KafkaUtil.disconnect();

			webResult.msg = op + " success";

			return;
		}

		if (OP_CLIENT_INFO.equals(op)) {

			KafkaClient kafkaClient = KafkaUtil.getClient();
			
			
			webResult.msg = kafkaClient.server;

			return;
		}
		
		
		
		
	}
	
	public static void info(HttpServletRequest request, WebResult webResult) throws Exception {
		
		try{
		KafkaClusterInfo clusterInfo = KafkaUtil.getClusterInfo();
		Map<String, TopicListing> topics  = KafkaUtil.listTopics(true);
		ConsumerGroupsInfo  consumerGroupsInfo = KafkaUtil.listConsumerGroups();
		
		if(consumerGroupsInfo!=null){
			//
		    List<ConsumerGroupDTO> consumerGroupDTOList = consumerGroupsInfo.consumerGroupDTOList;
		    webResult.consumerGroupDTOList = consumerGroupDTOList;
		}


		List<KafkaBroker> brokers = clusterInfo.brokers;
		Node controller = clusterInfo.controller;
		
		
		webResult.clusterInfo = clusterInfo;
		webResult.brokers = brokers;
		webResult.controller = controller;
		webResult.consumerGroupsInfo = consumerGroupsInfo;
		webResult.topics = topics;

		
		String zookeeperConnect = null;
		if(controller!=null){
			List<ConfigEntry> controllerConfig = KafkaUtil.getBrokerConfig(controller.id());
			zookeeperConnect = KafkaUtil.getZookeeperConnect(controllerConfig);
			webResult.zookeeperConnect = zookeeperConnect;

		}
		
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "kafka info error," + e;
		}

	}

	
	public static void getConfig(HttpServletRequest request, WebResult webResult) throws Exception {
		try {
			
			String type = WebUtil.get(request, "type");
			if (StringUtils.isBlank(type)) {
				webResult.msg = "config type blank";
				webResult.error = true;

				return;
			}
			
			if (!CONFIG_TYPES.contains(type)) {
				webResult.msg = "config type error";
				webResult.error = true;
				return;
			}
			
			webResult.configType = type;
			
			if(CONFIG_TYPE_BROKER.equals(type)){
				int brokerId = WebUtil.getInt(request, "brokerId", -1);
				if(brokerId<0){
					webResult.msg = "brokerId error";
					webResult.error = true;
					return;
				}
				webResult.brokerId = brokerId;
				webResult.configList = KafkaUtil.getBrokerConfig(brokerId);
				
				
				return;
			}
			
            if(CONFIG_TYPE_TOPIC.equals(type)){
            	
            	String topic = WebUtil.get(request, "name");
    			if (StringUtils.isBlank(topic)) {
    				webResult.msg = "topic blank";
    				webResult.error = true;

    				return;
    			}
    			
    			webResult.topic = topic;
				webResult.configList = KafkaUtil.getTopicConfig(topic);

				
				return;
			}
			
			
			
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "getConfig error," + e;
		}
	}
	
	public static void getLogDirInfo(HttpServletRequest request, WebResult webResult) throws Exception {
		int brokerId = WebUtil.getInt(request, "brokerId", -1);
		if(brokerId<0){
			webResult.msg = "brokerId error";
			webResult.error = true;
			return;
		}
		try{
			webResult.brokerId = brokerId;
			
			List<KafkaTopicReplicaSummaryInfo> topicReplicaSummaryInfoList = KafkaUtil.getBrokerLogDirInfoList(brokerId);
			webResult.topicReplicaSummaryInfoList = topicReplicaSummaryInfoList;
			
			
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "getLogDirInfo error," + e;
		}
		
	}


	public static void getTopicDesc(HttpServletRequest request, WebResult webResult) throws Exception {
		String topic = WebUtil.get(request, "topic");
		if (StringUtils.isBlank(topic)) {
			webResult.msg = "topic blank";
			webResult.error = true;

			return;
		}
		webResult.topic = topic;
		
		try{
			
			TopicDescription desc = KafkaUtil.getTopicDesc(topic);
			webResult.topicDescription = desc;
			
			
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "getTopicDesc error," + e;
		}
		
		
		
	}
	
	public static void getConsumerGroupDesc(HttpServletRequest request, WebResult webResult) throws Exception {
		String groupId = WebUtil.get(request, "groupId");
		if (StringUtils.isBlank(groupId)) {
			webResult.msg = "groupId blank";
			webResult.error = true;

			return;
		}
		webResult.groupId = groupId;
		
		try{
			
			ConsumerGroupDescription desc = KafkaUtil.getConsumerGroupDescription(groupId);
			webResult.consumerGroupDescription = desc;
			
			
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "getConsumerGroupDesc error," + e;
		}
		
	}
	
	public static void getOffsetInfos(HttpServletRequest request, WebResult webResult) throws Exception {
		String groupId = WebUtil.get(request, "groupId");
		if (StringUtils.isBlank(groupId)) {
			webResult.msg = "groupId blank";
			webResult.error = true;

			return;
		}
		webResult.groupId = groupId;
		
		try{		
			List<KafkaOffsetInfo> list = KafkaUtil.getOffsetInfos(groupId);		
			webResult.kafkaOffsetInfoList = list;		
		} catch (Throwable e) {
			webResult.error = true;
			webResult.msg = "getOffsetInfos error," + e;
		}
		
	}



}
